package com.openbigdata.gaussdb.doris;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.types.Row;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;

import java.sql.*;
import java.time.LocalDateTime;
import java.math.BigDecimal;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MySQLToGaussDBFullSync {
    private static final Logger LOG = LoggerFactory.getLogger(MySQLToGaussDBFullSync.class);
    
    // MySQL配置
    private static final String MYSQL_CONFIG = "jdbc:mysql://localhost:port/db?useSSL=false";
    private static final String MYSQL_USER = "user";
    private static final String MYSQL_PASSWORD = "password";

    // GaussDB配置
    private static final String GAUSS_CONFIG = "jdbc:postgresql://localhost:port/db";
    private static final String GAUSS_USER = "user";
    private static final String GAUSS_PASSWORD = "password";
    
    // 表名配置
    private static final String MYSQL_REPLY_TABLE = "reply";
    private static final String MYSQL_CREDIT_FACILITY_TABLE = "credit_facility";
    private static final String GAUSS_ODS_TABLE = "jiang.ods_data";
    private static final String GAUSS_DWD_TABLE = "jiang.dwd_data";
    private static final String GAUSS_ADS_TABLE = "jiang.ads_data";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1. MySQL到ODS层
        DataStream<Row> sourceData = env.addSource(new MySQLSourceFunction(
            MYSQL_CONFIG, MYSQL_USER, MYSQL_PASSWORD, MYSQL_REPLY_TABLE
        ));

        sourceData.addSink(JdbcSink.sink(
            "MERGE INTO " + GAUSS_ODS_TABLE + " t " +
            "USING (VALUES (?, ?, ?, ?, ?, ?)) AS s " +
            "(create_time, update_time, credit_amount, irr, period, credit_facility_id) " +
            "ON t.credit_facility_id = s.credit_facility_id " +
            "WHEN MATCHED THEN " +
            "  UPDATE SET create_time = s.create_time, " +
            "             update_time = s.update_time, " +
            "             credit_amount = s.credit_amount, " +
            "             irr = s.irr, " +
            "             period = s.period " +
            "WHEN NOT MATCHED THEN " +
            "  INSERT (create_time, update_time, credit_amount, irr, period, credit_facility_id) " +
            "  VALUES (s.create_time, s.update_time, s.credit_amount, s.irr, s.period, s.credit_facility_id)",
            (ps, row) -> {
                try {
                    ps.setTimestamp(1, Timestamp.valueOf((LocalDateTime) row.getField(0)));
                    ps.setTimestamp(2, Timestamp.valueOf((LocalDateTime) row.getField(1)));
                    ps.setBigDecimal(3, (BigDecimal) row.getField(2));
                    ps.setBigDecimal(4, (BigDecimal) row.getField(3));
                    ps.setInt(5, (Integer) row.getField(4));
                    ps.setString(6, (String) row.getField(5));
                    LOG.debug("Writing to GaussDB: credit_facility_id={}, credit_amount={}", 
                        row.getField(5), row.getField(2));
                } catch (Exception e) {
                    LOG.error("Error writing to GaussDB", e);
                    throw e;
                }
            },
            JdbcExecutionOptions.builder()
                .withBatchSize(1)
                .withMaxRetries(3)
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(GAUSS_CONFIG)
                .withDriverName("org.postgresql.Driver")
                .withUsername(GAUSS_USER)
                .withPassword(GAUSS_PASSWORD)
                .build()
        ));

        // 2. ODS到DWD层
        DataStream<Row> gaussData = env.addSource(new GaussDBSourceFunction(
            GAUSS_CONFIG, GAUSS_USER, GAUSS_PASSWORD, GAUSS_ODS_TABLE
        ));

        DataStream<Row> filteredData = gaussData.filter(row -> {
            BigDecimal creditAmount = (BigDecimal) row.getField(2);
            return creditAmount.compareTo(new BigDecimal("500000")) >= 0;
        }).map(new MapFunction<Row, Row>() {
            @Override
            public Row map(Row row) {
                Row newRow = new Row(3);
                newRow.setField(0, row.getField(1));  // update_time
                newRow.setField(1, row.getField(2));  // credit_amount
                newRow.setField(2, row.getField(5));  // credit_facility_id
                return newRow;
            }
        });

        filteredData.addSink(JdbcSink.sink(
            "INSERT INTO " + GAUSS_DWD_TABLE + 
            " (update_time, credit_amount, credit_facility_id) VALUES (?, ?, ?)",
            (ps, row) -> {
                ps.setTimestamp(1, Timestamp.valueOf((LocalDateTime) row.getField(0)));
                ps.setBigDecimal(2, (BigDecimal) row.getField(1));
                ps.setString(3, (String) row.getField(2));
            },
            JdbcExecutionOptions.builder()
                .withBatchSize(1)
                .withMaxRetries(3)
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(GAUSS_CONFIG)
                .withDriverName("org.postgresql.Driver")
                .withUsername(GAUSS_USER)
                .withPassword(GAUSS_PASSWORD)
                .build()
        ));

        // 3. DWD到ADS层
        DataStream<Row> dwdData = env.addSource(new DwdSourceFunction(
            GAUSS_CONFIG, GAUSS_USER, GAUSS_PASSWORD, GAUSS_DWD_TABLE
        ));

        DataStream<Row> creditFacilityData = env.addSource(new CreditFacilitySourceFunction(
            MYSQL_CONFIG, MYSQL_USER, MYSQL_PASSWORD, MYSQL_CREDIT_FACILITY_TABLE
        ));

        DataStream<Tuple2<String, String>> dwdStream = dwdData
            .map(new MapFunction<Row, Tuple2<String, String>>() {
                @Override
                public Tuple2<String, String> map(Row row) {
                    return new Tuple2<>(
                        row.getField(2).toString(),  // credit_facility_id
                        ""  // 占位，用于关联
                    );
                }
            })
            .returns(Types.TUPLE(Types.STRING, Types.STRING));

        DataStream<Tuple2<String, String>> creditFacilityStream = creditFacilityData
            .map(new MapFunction<Row, Tuple2<String, String>>() {
                @Override
                public Tuple2<String, String> map(Row row) {
                    return new Tuple2<>(
                        row.getField(0).toString(),  // id
                        row.getField(1).toString()   // lease_organization
                    );
                }
            })
            .returns(Types.TUPLE(Types.STRING, Types.STRING));

        DataStream<Row> joinedStream = dwdStream
            .join(creditFacilityStream)
            .where(new KeySelector<Tuple2<String, String>, String>() {
                @Override
                public String getKey(Tuple2<String, String> value) {
                    return value.f0;
                }
            })
            .equalTo(new KeySelector<Tuple2<String, String>, String>() {
                @Override
                public String getKey(Tuple2<String, String> value) {
                    return value.f0;
                }
            })
            .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            .apply(new JoinFunction<Tuple2<String, String>, Tuple2<String, String>, Row>() {
                @Override
                public Row join(Tuple2<String, String> dwd, Tuple2<String, String> cf) {
                    Row row = new Row(2);
                    row.setField(0, cf.f1);              // lease_organization
                    row.setField(1, dwd.f0);             // credit_facility_id
                    return row;
                }
            });

        joinedStream.addSink(JdbcSink.sink(
            "INSERT INTO " + GAUSS_ADS_TABLE + 
            " (lease_organization, credit_facility_id) VALUES (?, ?)",
            (ps, row) -> {
                ps.setString(1, (String) row.getField(0));
                ps.setString(2, (String) row.getField(1));
            },
            JdbcExecutionOptions.builder()
                .withBatchSize(1)
                .withMaxRetries(3)
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(GAUSS_CONFIG)
                .withDriverName("org.postgresql.Driver")
                .withUsername(GAUSS_USER)
                .withPassword(GAUSS_PASSWORD)
                .build()
        ));

        env.execute("MySQL to GaussDB Real-time Sync and Monitor");
    }

    // MySQL Source Function
    private static class MySQLSourceFunction implements SourceFunction<Row> {
        private volatile boolean isRunning = true;
        private final String url;
        private final String username;
        private final String password;
        private final String tableName;
        private Long lastId = 0L;

        public MySQLSourceFunction(String url, String username, String password, String tableName) {
            this.url = url;
            this.username = username;
            this.password = password;
            this.tableName = tableName;
        }

        @Override
        public void run(SourceContext<Row> ctx) throws Exception {
            Connection connection = DriverManager.getConnection(url, username, password);
            LOG.info("MySQL connection established");
            
            while (isRunning) {
                String sql = String.format(
                    "SELECT r.id, r.create_time, r.update_time, r.credit_amount, r.irr, r.period, r.credit_facility_id " +
                    "FROM %s r WHERE r.id > ? ORDER BY r.id LIMIT 100",
                    tableName
                );
                LOG.debug("Executing MySQL query: {}", sql);

                try (PreparedStatement stmt = connection.prepareStatement(sql)) {
                    stmt.setLong(1, lastId);
                    ResultSet rs = stmt.executeQuery();
                    boolean hasData = false;

                    while (rs.next()) {
                        hasData = true;
                        Row row = new Row(6);
                        row.setField(0, rs.getTimestamp("create_time").toLocalDateTime());
                        row.setField(1, rs.getTimestamp("update_time").toLocalDateTime());
                        row.setField(2, rs.getBigDecimal("credit_amount"));
                        row.setField(3, rs.getBigDecimal("irr"));
                        row.setField(4, rs.getInt("period"));
                        row.setField(5, rs.getString("credit_facility_id"));
                        
                        lastId = rs.getLong("id");
                        LOG.debug("Processing MySQL record: id={}, credit_facility_id={}", 
                            lastId, row.getField(5));
                        ctx.collect(row);
                    }

                    if (!hasData) {
                        LOG.debug("No new data found in MySQL, waiting...");
                        Thread.sleep(5000);
                    }
                } catch (Exception e) {
                    LOG.error("Error processing MySQL data", e);
                    throw e;
                }
            }
            
            connection.close();
            LOG.info("MySQL connection closed");
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }

    // GaussDB Source Function
    private static class GaussDBSourceFunction extends RichSourceFunction<Row> {
        private volatile boolean isRunning = true;
        private final String url;
        private final String username;
        private final String password;
        private final String tableName;
        private transient Connection connection;
        private LocalDateTime lastProcessedTime = null;

        public GaussDBSourceFunction(String url, String username, String password, String tableName) {
            this.url = url;
            this.username = username;
            this.password = password;
            this.tableName = tableName;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection(url, username, password);
        }

        @Override
        public void run(SourceContext<Row> ctx) throws Exception {
            while (isRunning) {
                String sql;
                if (lastProcessedTime == null) {
                    sql = String.format(
                        "SELECT create_time, update_time, credit_amount, irr, period, credit_facility_id " +
                        "FROM %s ORDER BY update_time",
                        tableName
                    );
                } else {
                    sql = String.format(
                        "SELECT create_time, update_time, credit_amount, irr, period, credit_facility_id " +
                        "FROM %s WHERE update_time > ? ORDER BY update_time",
                        tableName
                    );
                }

                try (PreparedStatement stmt = connection.prepareStatement(sql)) {
                    if (lastProcessedTime != null) {
                        stmt.setTimestamp(1, Timestamp.valueOf(lastProcessedTime));
                    }

                    ResultSet rs = stmt.executeQuery();
                    boolean hasData = false;
                    
                    while (rs.next()) {
                        hasData = true;
                        Row row = new Row(6);
                        row.setField(0, rs.getTimestamp("create_time").toLocalDateTime());
                        row.setField(1, rs.getTimestamp("update_time").toLocalDateTime());
                        row.setField(2, rs.getBigDecimal("credit_amount"));
                        row.setField(3, rs.getBigDecimal("irr"));
                        row.setField(4, rs.getInt("period"));
                        row.setField(5, rs.getString("credit_facility_id"));
                        
                        lastProcessedTime = rs.getTimestamp("update_time").toLocalDateTime();
                        ctx.collect(row);
                    }

                    if (!hasData) {
                        Thread.sleep(5000);
                    }
                }
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }

        @Override
        public void close() throws Exception {
            if (connection != null && !connection.isClosed()) {
                connection.close();
            }
        }
    }

    // DWD Source Function
    private static class DwdSourceFunction extends RichSourceFunction<Row> {
        private volatile boolean isRunning = true;
        private final String url;
        private final String username;
        private final String password;
        private final String tableName;
        private transient Connection connection;
        private LocalDateTime lastProcessedTime = null;

        public DwdSourceFunction(String url, String username, String password, String tableName) {
            this.url = url;
            this.username = username;
            this.password = password;
            this.tableName = tableName;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection(url, username, password);
        }

        @Override
        public void run(SourceContext<Row> ctx) throws Exception {
            while (isRunning) {
                String sql;
                if (lastProcessedTime == null) {
                    sql = String.format(
                        "SELECT update_time, credit_amount, credit_facility_id " +
                        "FROM %s ORDER BY update_time",
                        tableName
                    );
                } else {
                    sql = String.format(
                        "SELECT update_time, credit_amount, credit_facility_id " +
                        "FROM %s WHERE update_time > ? ORDER BY update_time",
                        tableName
                    );
                }

                try (PreparedStatement stmt = connection.prepareStatement(sql)) {
                    if (lastProcessedTime != null) {
                        stmt.setTimestamp(1, Timestamp.valueOf(lastProcessedTime));
                    }

                    ResultSet rs = stmt.executeQuery();
                    boolean hasData = false;
                    
                    while (rs.next()) {
                        hasData = true;
                        Row row = new Row(3);
                        row.setField(0, rs.getTimestamp("update_time").toLocalDateTime());
                        row.setField(1, rs.getBigDecimal("credit_amount"));
                        row.setField(2, rs.getString("credit_facility_id"));
                        
                        lastProcessedTime = rs.getTimestamp("update_time").toLocalDateTime();
                        ctx.collect(row);
                    }

                    if (!hasData) {
                        Thread.sleep(5000);
                    }
                }
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }

        @Override
        public void close() throws Exception {
            if (connection != null && !connection.isClosed()) {
                connection.close();
            }
        }
    }

    // Credit Facility Source Function
    private static class CreditFacilitySourceFunction extends RichSourceFunction<Row> {
        private volatile boolean isRunning = true;
        private final String url;
        private final String username;
        private final String password;
        private final String tableName;
        private transient Connection connection;
        private Long lastId = 0L;

        public CreditFacilitySourceFunction(String url, String username, String password, String tableName) {
            this.url = url;
            this.username = username;
            this.password = password;
            this.tableName = tableName;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection(url, username, password);
        }

        @Override
        public void run(SourceContext<Row> ctx) throws Exception {
            while (isRunning) {
                String sql;
                if (lastId == 0L) {
                    sql = String.format(
                        "SELECT id, lease_organization FROM %s ORDER BY id",
                        tableName
                    );
                } else {
                    sql = String.format(
                        "SELECT id, lease_organization FROM %s WHERE id > ? ORDER BY id",
                        tableName
                    );
                }

                try (PreparedStatement stmt = connection.prepareStatement(sql)) {
                    if (lastId > 0) {
                        stmt.setLong(1, lastId);
                    }

                    ResultSet rs = stmt.executeQuery();
                    boolean hasData = false;
                    
                    while (rs.next()) {
                        hasData = true;
                        Row row = new Row(2);
                        row.setField(0, rs.getString("id"));
                        row.setField(1, rs.getString("lease_organization"));
                        
                        lastId = rs.getLong("id");
                        ctx.collect(row);
                    }

                    if (!hasData) {
                        Thread.sleep(5000);
                    }
                }
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }

        @Override
        public void close() throws Exception {
            if (connection != null && !connection.isClosed()) {
                connection.close();
            }
        }
    }
} 